package com.clp.inflxudb.core;

import com.clp.inflxudb.core.domain.*;
import com.clp.inflxudb.core.enums.*;

import java.time.Instant;
import java.util.List;

class StrFlux implements Flux{
    private static final int NONE = 0x01;
    private static final int BUCKET = 0x02;
    private static final int RANGE = 0x04;
    private static final int FILTER = 0x08;
    private static final int YIELD = 0x10;

    private final String name; // 变量名
    private final StringBuilder sb;
    private int state;

    private boolean isPivoted; // 是否执行了pivot()

    public StrFlux() {
        this.name = null;
        this.sb = new StringBuilder("data = ");
        this.state = NONE;
        this.isPivoted = false;
    }

    public StrFlux(String name) {
        this.name = name;
        this.sb = new StringBuilder();
        this.state = NONE;
        this.isPivoted = false;
    }

    /**
     * 检查方法执行顺序的正确性
     *
     * @param states：状态列表
     */
    private boolean checkState(int states) {
        return (this.state & states) != 0;
    }

    private void checkStateAfterRangeBeforeYield() {
        if (state < RANGE) {
            throw new RuntimeException("请先执行range()方法！");
        }
        if (state >= YIELD) {
            throw new RuntimeException("该Flux语句已经结束！");
        }
    }

    /**
     * 第1步：指定bucketName
     *
     * @param bucketName：桶名称
     * @return：自身
     */
    @Override
    public StrFlux bucket(String bucketName) {
        if (!checkState(NONE)) throw new RuntimeException("已执行bucket()方法！");
        sb.append("from(bucket: \"").append(bucketName).append("\")");
        state = BUCKET;
        return this;
    }

    /**
     * 第2步：指定range()
     *
     * @param startTime：开始时间
     * @param stopTime：结束时间
     * @param tsUnit：时间单位
     * @return：自身
     */
    @Override
    public StrFlux range(int startTime, int stopTime, TsUnit tsUnit) {
        if (!checkState(BUCKET)) throw new RuntimeException("请先执行bucket()方法！");
        sb.append("\n\t|> range(" + "start: ").append(startTime).append(tsUnit.getCh()).append(", stop: ")
                .append(stopTime).append(tsUnit.getCh()).append(")");
        state = RANGE;
        return this;
    }

    @Override
    public StrFlux range(Instant startTime, Instant stopTime) {
        if (!checkState(BUCKET)) throw new RuntimeException("请先执行bucket()方法！");
        sb.append("\n\t|> range(" + "start: ").append(startTime.toString());
        if (stopTime != null) {
            sb.append(", stop: ").append(stopTime);
        }
        sb.append(")");
        state = RANGE;
        return this;
    }

    @Override
    public Flux range(TsRange tsRange) {
        if (tsRange instanceof InstantTsRange) {
            InstantTsRange instantTsRange = (InstantTsRange) tsRange;
            return range(instantTsRange.getStartTime(), instantTsRange.getStopTime());
        } else if (tsRange instanceof IntTsRange) {
            IntTsRange intTsRange = (IntTsRange) tsRange;
            return range(intTsRange.getStartTime(), intTsRange.getStopTime(), intTsRange.getTsUnit());
        } else {
            return this;
        }
    }

    private static String castToString(Object valueObj) {
        if (valueObj == null) {
            return null;
        }
        String valueStr;
        if (valueObj instanceof String) {
            valueStr = "\"" + valueObj + "\"";
        } else {
            valueStr = valueObj.toString();
        }
        return valueStr;
    }

    /**
     * 过滤列表属性判别
     *
     * @param fluxKVList：键值对列表
     * @param logicOp：逻辑连接
     * @return：自身
     */
    @Override
    public StrFlux filter(List<FluxKV> fluxKVList, LogicOp logicOp) {
        checkStateAfterRangeBeforeYield();
        if (fluxKVList.isEmpty()) {
            return this;
        }

        FluxKV attr = fluxKVList.get(0);
        String keyStr = attr.getKey();
        String valueStr = castToString(attr.getValue());
        sb.append("\n\t|> filter(fn: (r) => r[\"").append(keyStr).append("\"] ").append(attr.getCompareOp()
                .getOpStr()).append(" ").append(valueStr);
        for (int i = 1; i < fluxKVList.size(); i++) {
            attr = fluxKVList.get(i);
            keyStr = attr.getKey();
            valueStr = castToString(attr.getValue());
            sb.append(" ").append(logicOp.getOpStr()).append(" r[\"").append(keyStr).append("\"] ")
                    .append(attr.getCompareOp().getOpStr()).append(" ").append(valueStr);
        }
        sb.append(")");
        state = FILTER;
        return this;
    }

    @Override
    public Flux filter(FluxKVGroup fluxKVGroup) {
        checkStateAfterRangeBeforeYield();
        if (fluxKVGroup == null) return this;
        sb.append("\n\t|> filter(fn: (r) => ").append(postOrderFluxKVGroup(fluxKVGroup)).append(")");
        state = FILTER;
        return this;
    }

    /**
     * 后序遍历该键值对组（二叉树），从而递归产生判别式
     * @param fluxKVGroup：键值对组
     * @return：转换后的条件判断字符串
     */
    private String postOrderFluxKVGroup(FluxKVGroup fluxKVGroup) {
        if (fluxKVGroup.isLeaf()) {
            FluxKV fluxKV = fluxKVGroup.getFluxKV();
            String keyStr = fluxKV.getKey();
            String valueStr = castToString(fluxKV.getValue());
            switch (fluxKV.getKvType()) {
                case TAG:
                    return "r[\"" + keyStr + "\"] == " + valueStr;
                case FIELD:
                    return "(r[\"" + FluxKey._field + "\"] == \"" + keyStr + "\" and r[\"" + FluxKey._value + "\"] == " + valueStr + ")";
                case TIMESTAMP:
                    return "r[\"" + FluxKey._time + "\"] == " + valueStr;
                default:
                    throw new RuntimeException("非法字段条件");
            }
        } else if (fluxKVGroup.hasLeftGroup() && fluxKVGroup.hasRightGroup()) {
            FluxKVGroup leftGroup = fluxKVGroup.getLeftGroup();
            FluxKVGroup rightGroup = fluxKVGroup.getRightGroup();
            LogicOp logicOp = fluxKVGroup.getLogicOp();
            return "(" + postOrderFluxKVGroup(leftGroup) + " "
                    + logicOp.getOpStr() + " " + postOrderFluxKVGroup(rightGroup) + ")";
        } else if (fluxKVGroup.hasLeftGroup()) {
            FluxKVGroup leftGroup = fluxKVGroup.getLeftGroup();
            return "(" + postOrderFluxKVGroup(leftGroup) + ")";
        } else {
            FluxKVGroup rightGroup = fluxKVGroup.getRightGroup();
            return "(" + postOrderFluxKVGroup(rightGroup) + ")";
        }
    }

    /**
     * 过滤列表属性判别
     *
     * @param fluxKVList：键值对列表
     * @return：自身
     */
    @Override
    public StrFlux filterAnd(List<FluxKV> fluxKVList) {
        return filter(fluxKVList, LogicOp.and);
    }

    /**
     * 过滤列表属性判别
     *
     * @param fluxKVList：键值对列表
     * @return：自身
     */
    @Override
    public StrFlux filterOr(List<FluxKV> fluxKVList) {
        return filter(fluxKVList, LogicOp.or);
    }

    /**
     * 过滤单个属性判别
     * @param keyStr：键
     * @param valueObj：值对象
     * @param compareOp：值的比较
     * @return：自身
     */
    @Override
    public Flux filter(String keyStr, Object valueObj, CompareOp compareOp) {
        return filter(new FluxKV(keyStr, valueObj, CompareOp.EQUAL));
    }

    /**
     * 过滤单个属性判别
     *
     * @param fluxKV：键值对
     * @return：自身
     */
    @Override
    public StrFlux filter(FluxKV fluxKV) {
        checkStateAfterRangeBeforeYield();
        if (fluxKV == null) return this;
        String keyStr = fluxKV.getKey();
        String valueStr = castToString(fluxKV.getValue());

        switch (fluxKV.getKvType()) {
            case FIELD:
//                sb.append("\n\t|> filter(fn: (r) => r[\"_field\"] = \"" + keyStr + "\" and r[\"_value\"] = " + valueStr + ")");
//                break;
            case TAG:
                sb.append("\n\t|> filter(fn: (r) => r[\"").append(keyStr).append("\"] ").append(fluxKV.getCompareOp()
                        .getOpStr()).append(" ").append(valueStr).append(")");
//                break;
                break;
            case TIMESTAMP:
                sb.append("\n\t|> filter(fn: (r) => r[\"_time\"] = ").append(valueStr).append(")");
                break;
        }
        state = FILTER;
        return this;
    }

    @Override
    public Flux filterCustom(String queryStr) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> filter(fn: (r) => ").append(queryStr).append(")");
        state = FILTER;
        return this;
    }

    @Override
    public Flux window(int every, TsUnit tsUnit) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> window(every: ").append(every).append(tsUnit.getCh()).append(")");
        return this;
    }

    @Override
    public Flux window(int every, TsUnit tsUnit, int offset, TsUnit offsetTsUnit) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> window(every: ").append(every).append(tsUnit.getCh()).append(", offset: ").append(offset)
                .append(offsetTsUnit.getCh()).append(")");
        return this;
    }

    @Override
    public Flux windowInf() {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> window(every: inf)");
        return this;
    }

    @Override
    public Flux aggregateWindow(int every, TsUnit tsUnit, Func func) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> window(every: ").append(every).append(tsUnit.getCh()).append(", fn: ").append(func.getVal())
                .append(")");
        return this;
    }

    @Override
    public Flux mean() {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> mean()");
        return this;
    }

    @Override
    public Flux mean(String columnName) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> mean(column: \"").append(columnName).append("\")");
        return this;
    }

    @Override
    public Flux min() {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> min()");
        return this;
    }

    @Override
    public Flux min(String columnName) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> min(column: \"").append(columnName).append("\")");
        return this;
    }

    @Override
    public Flux max() {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> max()");
        return this;
    }

    @Override
    public Flux max(String columnName) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> max(column: \"").append(columnName).append("\")");
        return this;
    }

    @Override
    public Flux sum() {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> sum()");
        return this;
    }

    @Override
    public Flux sum(String columnName) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> sum(column: \"").append(columnName).append("\")");
        return this;
    }

    @Override
    public Flux group(List<String> keyStrList, GroupMode groupMode) {
        checkStateAfterRangeBeforeYield();
        if (keyStrList.isEmpty()) return this;
        sb.append("\n\t|> group(columns: [\"").append(keyStrList.get(0)).append("\"");
        for (int i = 1; i < keyStrList.size(); i++) {
            sb.append(", \"").append(keyStrList.get(i)).append("\"");
        }
        sb.append("], mode: \"").append(groupMode.getVal()).append("\")");
        return this;
    }

    @Override
    public Flux duplicate(String fromColumnName, String toColumnName) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> duplicate(column: \"").append(fromColumnName).append("\", as: \"").append(toColumnName)
                .append("\")");
        return this;
    }

    @Override
    public Flux drop(List<String> keyStrList) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> drop(columns: [\"").append(keyStrList.get(0)).append("\"");
        for (int i = 1; i < keyStrList.size(); i++) {
            sb.append(", \"").append(keyStrList.get(i)).append("\"");
        }
        sb.append("])");
        return this;
    }

    @Override
    public Flux sort(boolean desc) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> sort(desc: ").append(desc).append(")");
        return this;
    }

    @Override
    public Flux sort(List<String> keyStrList, boolean desc) {
        if (keyStrList == null || keyStrList.isEmpty()) return sort(desc);
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> sort(columns: [\"").append(keyStrList.get(0)).append("\"");
        for (int i = 1; i < keyStrList.size(); i++) {
            sb.append(", \"").append(keyStrList.get(i)).append("\"");
        }
        sb.append("], desc: ").append(desc).append(")");
        return this;
    }

    @Override
    public Flux limit(int offset, int num) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> limit(n:").append(num).append(", offset: ").append(offset).append(")");
        return this;
    }

    @Override
    public Flux map(String tblName, String keyStr, String execution) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> map(fn: (").append(tblName).append(") => ({ ").append(tblName).append(" with ")
                .append(keyStr).append(": ").append(execution).append(" }))");
        return this;
    }

    @Override
    public Flux map(String tblName, String execution) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> map(fn: (").append(tblName).append(") => ({").append(execution).append("}))");
        return this;
    }

    @Override
    public Flux pivot(String rowKey, String columnKey, String valueColumn) {
        if (isPivoted) throw new RuntimeException("已经执行了一次pivot()操作！");
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> pivot(rowKey: [\"").append(rowKey).append("\"], columnKey: [\"").append(columnKey)
                .append("\"], valueColumn: \"").append(valueColumn).append("\")");
        isPivoted = true;
        return this;
    }

    @Override
    public Flux pivot() {
        return pivot(FluxKey._time, FluxKey._field, FluxKey._value);
    }

    @Override
    public boolean isPivoted() {
        return isPivoted;
    }

    @Override
    public Flux increase() {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> increase()");
        return this;
    }

    @Override
    public Flux movingAverage(int num) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> movingAverage(n: ").append(num).append(")");
        return this;
    }

    @Override
    public Flux timedMovingAverage(int every, TsUnit everyTsUnit, int period, TsUnit periodTsUnit) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> timedMovingAverage(every: ").append(every).append(everyTsUnit.getCh()).append(", period: ")
                .append(period).append(periodTsUnit.getCh()).append(")");
        return this;
    }

    @Override
    public Flux derivative(int unit, TsUnit tsUnit, boolean nonNegative) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> derivative(unit: ").append(unit).append(tsUnit.getCh()).append(", nonNegative: ")
                .append(nonNegative).append(")");
        return this;
    }

    @Override
    public Flux aggregateRate(int every, TsUnit everyTsUnit, int unit, TsUnit unitTsUnit) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> aggregateRate(every: ").append(every).append(everyTsUnit.getCh()).append(", unit: ")
                .append(unit).append(unitTsUnit.getCh()).append(")");
        return this;
    }

    @Override
    public Flux fill(boolean usePrevious) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> fill(userPrevious: ").append(usePrevious).append(")");
        return this;
    }

    @Override
    public Flux fill(int value) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> fill(value: ").append(value).append(")");
        return this;
    }

    @Override
    public Flux median(MedianMethod medianMethod) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> median(method: \"").append(medianMethod.getVal()).append("\")");
        return this;
    }

    @Override
    public Flux quantile(double q, QuantileMethod quantileMethod) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> quantile(q: ").append(q).append(", method: \"").append(quantileMethod.getVal()).append("\")");
        return this;
    }

    @Override
    public Flux cumulativeSum() {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> cumulativeSum()");
        return this;
    }

    @Override
    public Flux first() {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> first()");
        return this;
    }

    @Override
    public Flux first(String columnName) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> first(column: \"").append(columnName).append("\")");
        return this;
    }

    @Override
    public Flux last() {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> last()");
        return this;
    }

    @Override
    public Flux last(String columnName) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> last(column: \"").append(columnName).append("\")");
        return this;
    }

    @Override
    public Flux tableFind(List<FluxKV> fluxKVList) {
        checkStateAfterRangeBeforeYield();

        FluxKV attr = fluxKVList.get(0);
        String keyStr = attr.getKey();
        String valueStr = castToString(attr.getValue());
        sb.append("\n\t|> tableFind(fn: (key) => key.").append(keyStr).append(" ").append(attr.getCompareOp()
                .getOpStr()).append(" ").append(valueStr);
        for (int i = 1; i < fluxKVList.size(); i++) {
            attr = fluxKVList.get(i);
            keyStr = attr.getKey();
            valueStr = castToString(attr.getValue());
            sb.append(" and key.").append(keyStr).append(" ").append(attr.getCompareOp().getOpStr()).append(" ")
                    .append(valueStr);
        }
        sb.append(")");
        return this;
    }

    @Override
    public Flux getColumn(String keyStr) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> getColumn(column: \"").append(keyStr).append("\")");
        return this;
    }

    @Override
    public Flux getRecord(int index) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> getRecord(idx: ").append(index).append(")");
        return this;
    }

    @Override
    public Flux truncateTimeColumn(int unit, TsUnit tsUnit) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> truncateTimeColumn(unit: ").append(unit).append(tsUnit.getCh()).append(")");
        return this;
    }

    @Override
    public Flux count(String keyStr) {
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> count(column: \"").append(keyStr).append("\")");
        return this;
    }

    @Override
    public Flux customClause(String str) {
        checkStateAfterRangeBeforeYield();
        sb.append(str);
        return this;
    }

    /**
     * 结束
     *
     * @param name：名称
     * @return：自身
     */
    @Override
    public String yield(String name) {
        if (checkState(YIELD)) return sb.toString();
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> yield(name: \"").append(name).append("\")");
        state = YIELD;
        return sb.toString();
    }

    /**
     * 结束
     *
     * @return：自身
     */
    @Override
    public String yield() {
        if (checkState(YIELD)) return sb.toString();
        checkStateAfterRangeBeforeYield();
        sb.append("\n\t|> yield()");
        state = YIELD;
        return sb.toString();
    }
}
